package com.headfirst.dmp.report.ProvinceCityRpt

import java.util.Properties

import com.typesafe.config.{Config, ConfigFactory}
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * 统计各省市的数据量分布情况
  *
  * 数据基于刚才经过ETL形成的parquet文件  （tips:  parquet文件存和取都依赖于spark-sql）
  *
  * 需求 1：
  * 将统计出来的结果存储成JSON文件
  * 需求 2：
  * 将统计出来的结果存储到Mysql中 （本次存mysql中）
  */
object ProvinceCityRPT_Mysql {

  def main(args: Array[String]): Unit = {

    //1.判断参数个数
    if (args.length != 1) {
      print(
        """
          |com.hedfirst.dmp.tools.Bzip2ParquetV2
          |参数：
          |    logInputPath
        """.stripMargin)
      sys.exit(-1) //非正常退出
    }

    //2.创建sparksession对象
    val session: SparkSession = SparkSession.builder()
      .appName(s"${this.getClass.getSimpleName}")
      .master("local[*]")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") //设置序列化方式采用KryoSerializer方式（默认的是java序列化）
      .getOrCreate()

    //3.读取parquet文件
    val frame: DataFrame = session.read.parquet(args(0))

    //4.将DataFrame数据注册成视图
    frame.createTempView("v_log")

    //5.执行sql
    val sql: String = "select provincename,cityname,count(*) ct from v_log group by provincename,cityname order by provincename,cityname "
    val result: DataFrame = session.sql(sql)

    /**
      * 6.将数据输出到mysql中去
      */
    //此处使用工具类typesafe来读取配置文件
    //ConfigFactory.load()会自动加载application.conf/application.json/application.properties名称的配置文件
    val load: Config = ConfigFactory.load()
    val prop = new Properties()
    prop.setProperty("user", load.getString("jdbc.user"))
    prop.setProperty("password", load.getString("jdbc.password"))


    //也可以使用类加载器的方式读取配置文件
    //    val stream: InputStream = ProvinceCityRPT_Mysql.getClass.getClassLoader.getResourceAsStream("application.conf")
    //    prop.load(stream)
    //    prop.getProperty("user")
    //    prop.getProperty("password")

    result.write.mode(saveMode = "append").jdbc(load.getString("jdbc.url"), load.getString("jdbc.tableName"), prop)

    //关流
    session.stop()

  }

}
